package com.song.hbaseChap_5_01;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.net.TableMapping;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider.Text;

public class HBaseInputMapReduce extends Configured implements Tool{

	static final Log LOG = LogFactory.getLog(HBaseInputMapReduce.class);
	
	public static final String NAME ="Example Test1";
	
	public static final String TEMP_INDEX_PATH = "/tmp/example";
	public static final String inputTable ="example";
	
	public static void main(String[] args) {
		try {
			int ret = ToolRunner.run(new HBaseInputMapReduce(), args);
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public int run(String[] arg0) throws Exception {
		try {
			Configuration conf  =HBaseConfiguration.create();
			Scan scan = new Scan();
			scan.setBatch(0);
			scan.setCaching(10000);
			scan.setMaxVersions();
//			scan.setTimeRange(minStamp, maxStamp)
			scan.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("qulifier1"));
			
			conf.set("hbase.master", "macsong:60000");
			conf.set("hbase.zookeeper.quorum", "macsong");
			conf.setBoolean("mapred.map.tasks.speculative.execution", false);
			conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);
			
			Path tempIndexPath = new Path(TEMP_INDEX_PATH);
			
			try {
				FileSystem fs = FileSystem.get(conf);
				if(fs.exists(tempIndexPath))
				{
					fs.delete(tempIndexPath ,true);
				}
				
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			Job job = new Job(conf,NAME);
			TableMapReduceUtil.initTableMapperJob(inputTable, scan, ExampleMapper.class, Text.class, Text.class, job);
			job.setNumReduceTasks(0);
			job.setOutputFormatClass(TextOutputFormat.class);
			FileOutputFormat.setOutputPath(job, tempIndexPath);

			int success = job.waitForCompletion(true) ? 0:1;
			
			System.exit(success);
		} catch (IOException | ClassNotFoundException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return 0;
	}
	
	
	

	
	
}
